package com.lvmama.pangolin.timer.processor;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.log4j.Logger;
import org.quartz.JobExecutionContext;
import org.springframework.beans.factory.annotation.Autowired;

import com.lvmama.pangolin.common.Constant;
import com.lvmama.pangolin.common.RowkeyGenerator;
import com.lvmama.pangolin.common.StringTools;
import com.lvmama.pangolin.hbaseutil.HBaseUtils;
import com.lvmama.pangolin.timer.CommonJob;
import com.lvmama.pangolin.util.SpringContextUtil;


public class ExchangeCommodityData  extends CommonJob
{

	@Autowired
	private HBaseUtils hBaseUtils;
	
	private Logger logger = Logger.getLogger(ExchangeCommodityData.class);
	
	@Override
	public void runJob(JobExecutionContext context, String parameter)
	{
		logger.info("转换数据开始....");
		//保存产品信息
		if (hBaseUtils == null)
		{
			hBaseUtils = (HBaseUtils) SpringContextUtil
					.getBean(HBaseUtils.class);
		}
		
		//LMM数据转换
		String commodity = "competition_ticket_commodity_h_lmm";
		String commodity_new = "competition_ticket_commodity_h_lmm_new";
		String price = "competition_ticket_commodity_price_lmm";
		String price_new = "competition_ticket_commodity_price_lmm_new";
		
		String product = "competition_ticket_product_h_lmm";
		String product_new = "competition_ticket_product_h_lmm_new";
		
		logger.info("转换LMM数据开始....");
//		dealExchange(commodity_new,price_new,commodity,price);
		dealExchangeProd(product_new,product);
		logger.info("转换LMM数据开始....");
		
		//TC数据转换
		
		commodity = "competition_ticket_commodity_h_tc";
		commodity_new = "competition_ticket_commodity_h_tc_new";
		price = "competition_ticket_commodity_price_tc";
		price_new = "competition_ticket_commodity_price_tc_new";
		
		product = "competition_ticket_product_h_tc";
		product_new = "competition_ticket_product_h_tc_new";
		
		logger.info("转换TC数据开始....");
//		dealExchange(commodity_new,price_new,commodity,price);
		dealExchangeProd(product_new,product);
		logger.info("转换TC数据结束....");
		
		//TN数据转换
		commodity = "competition_ticket_commodity_h_tn";
		commodity_new = "competition_ticket_commodity_h_tn_new";
		price = "competition_ticket_commodity_price_tn";
		price_new = "competition_ticket_commodity_price_tn_new";
		
		product = "competition_ticket_product_h_tn";
		product_new = "competition_ticket_product_h_tn_new";
		
		logger.info("转换TN数据开始....");
//		dealExchange(commodity_new,price_new,commodity,price);
		dealExchangeProd(product_new,product);
		logger.info("转换TN数据结束....");
		
		//XC数据转换
		commodity = "competition_ticket_commodity_h_xc";
		commodity_new = "competition_ticket_commodity_h_xc_new";
		price = "competition_ticket_commodity_price_xc";
		price_new = "competition_ticket_commodity_price_xc_new";
		
		product = "competition_ticket_product_h_xc";
		product_new = "competition_ticket_product_h_xc_new";
		
		logger.info("转换XC数据开始....");
//		dealExchange(commodity_new,price_new,commodity,price);
		dealExchangeProd(product_new,product);
		logger.info("转换XC数据结束....");
		
		//QNR数据转换
		commodity = "competition_ticket_commodity_h_qnr";
		commodity_new = "competition_ticket_commodity_h_qnr_new";
		price = "competition_ticket_commodity_price_qnr";
		price_new = "competition_ticket_commodity_price_qnr_new";
		
		product = "competition_ticket_product_h_qnr";
		product_new = "competition_ticket_product_h_qnr_new";
		
		logger.info("转换QNR数据开始....");
//		dealExchange(commodity_new,price_new,commodity,price);
		dealExchangeProd(product_new,product);
		logger.info("转换QNR数据结束....");
		
		logger.info("转换数据结束....");
		
	}
	
	private void dealExchange(String commodityTableName,String priceTableName,String newCommTable,String newPriceTab)
	{
		try
		{
			List<Map<String,String>> result = hBaseUtils.queryAllAndRow(commodityTableName, Constant.HB_VARIABLE_FAMILY);
			if(result != null && !result.isEmpty() )
			{
				SimpleDateFormat format = new SimpleDateFormat("yyyyMMdd");
				Map<String,String> priceResult = null; 
				for (Map<String, String> map : result)
				{
					map.remove("rowkey");
					String dateStr = map.get("insert_time");
					if(!StringTools.isValid(dateStr))
					{
						dateStr = map.get("create_time");
					}
					Date  date = format.parse(dateStr);
					String newRowkey = RowkeyGenerator.generateRowkey(map.get("commodity_id"), 
							date, map.get("product_id"));
					
					hBaseUtils.save(map, newCommTable, Constant.HB_VARIABLE_FAMILY, newRowkey);
					
//					priceResult = hBaseUtils.queryNoVersion(newPriceTab, Constant.HB_VARIABLE_FAMILY,rowkey);
					priceResult = new HashMap<String,String>();
					priceResult.put("sale_price", map.get("sale_price"));
					priceResult.put("task_id", map.get("task_id"));
					priceResult.put("grab_time", dateStr);
					priceResult.put("sale_date", dateStr);
					priceResult.put("commodity_id", map.get("commodity_id"));
					hBaseUtils.save(priceResult, newPriceTab, Constant.HB_VARIABLE_FAMILY, newRowkey);
					
				}
				
			}
		}catch( Exception e)
		{
			logger.error("转换数据失败！",e);
		}
		
		
		
	}
	
	
	private void dealExchangeProd(String prodTableName,String newProdTab)
	{
		try
		{
			List<Map<String,String>> result = hBaseUtils.queryAll(prodTableName, Constant.HB_VARIABLE_FAMILY);
			if(result != null && !result.isEmpty() )
			{
				SimpleDateFormat format = new SimpleDateFormat("yyyyMMdd");
				for (Map<String, String> map : result)
				{
					String dateStr = map.get("insert_time");
					if(!StringTools.isValid(dateStr))
					{
						dateStr = map.get("create_time");
					}
					Date  date = format.parse(dateStr);
					String newRowkey = RowkeyGenerator.generateRowkey(map.get("product_id"), 
							date, map.get("task_id"));
					
					hBaseUtils.save(map, newProdTab, Constant.HB_VARIABLE_FAMILY, newRowkey);
					
				}
				
			}
		}catch( Exception e)
		{
			logger.error("转换数据失败！",e);
		}
	}
	
	public static void main(String[] args) throws ParseException
	{
		SimpleDateFormat format = new SimpleDateFormat("yyyyMMdd");
		String newRowkey = RowkeyGenerator.generateRowkey("3131200", 
				format.parse("20170612"), "181833");
		System.out.println(newRowkey);
	}
	
}
